package com.echo.im.server.ack;

import com.echo.im.common.enums.IMCmdType;
import com.echo.im.common.model.mq.OfflineMsgMqInfo;
import com.echo.im.mq.event.producer.EventPublisher;
import com.echo.im.mq.event.model.OfflineMessageEvent;
import com.echo.im.server.netty.context.UserChannelMap;
import io.netty.channel.Channel;
import io.netty.util.HashedWheelTimer;
import jakarta.annotation.PreDestroy;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;

/**
 *
 *  * 消息确认和超时处理管理器
 *  * 采用 HashedWheelTimer 进行高效的定时任务管理
 *  * 当消息超时时，将其加入到离线消息队列中
 *
 *
 * 消息确认等待队列管理器
 * 管理消息的ACK确认和超时处理
 * 采用HashedWheelTimer进行高效的定时任务管理
 * 当消息超时时，加入到离线消息队列中
 *
 * 现有代码已经通过null检查有效地处理了并发场景下可能的竞态条件，因此没有必要再增加同步块。
 * ConcurrentHashMap的线程安全特性与代码中的null检查结合起来，已经足够确保系统的正确性和稳定性。
 *
 * 简洁：无需额外的同步块，代码更为简洁、清晰。
 * 性能：避免了可能的锁竞争，从而提高了并发性能。
 *
 * 时间轮：
 * 精度与性能: 这将使时间轮在一圈内精确处理所有5秒延迟的任务，并且由于每个tick是500毫秒，系统消耗的资源较少。
 * 避免多圈转动: 因为每个任务都只有5秒的延迟，时间轮的配置可以确保任务在一圈内完成，无需等待多次轮转。
 * 适应性: 如果你之后有不同的延迟需求，例如3秒或7秒的延迟任务，可以调整 tickDuration 或 ticksPerWheel，不过对于当前5秒固定延迟任务来说，这种配置已经足够完美。
 *
 *
 * @author echo
 * @date 2024-08-15
 */
@RequiredArgsConstructor
@Component
@Slf4j
public class MessageAckManager {

    private final ConcurrentHashMap<Long, PendingAckInfo> pendingAckQueue = new ConcurrentHashMap<>();

    private static final HashedWheelTimer WHEEL_TIMLE = new HashedWheelTimer(500, TimeUnit.MILLISECONDS, 10);

    private final EventPublisher eventPublisher;

    private final OfflineMessageEvent offlineMessageEvent;

    @Data
    @AllArgsConstructor
    public static class PendingAckInfo {
        private Integer code;
        private Long userId;
        private String channelShortId;
    }

    public void addPendingAck(Long msgId, Long userId, String channelShortId, OfflineMsgMqInfo.OfflineMsgEnum type) {
        if (msgId == null) {
            log.warn("添加消息时接收到空的消息ID");
            return;
        }

        // 使用 computeIfAbsent 避免多线程并发导致的重复计算
        pendingAckQueue.computeIfAbsent(msgId, key -> new PendingAckInfo(type.code(), userId, channelShortId));
        log.debug("消息ID: {} 已添加到等待队列", msgId);

        // 调度超时任务，3秒后执行
        WHEEL_TIMLE.newTimeout(timeout -> handleAckTimeout(msgId), 3, TimeUnit.SECONDS);
    }

    public void confirmMessageAck(Long msgId) {
        if (msgId == null) {
            log.warn("确认消息时接收到空的消息ID");
            return;
        }

        PendingAckInfo ackInfo = pendingAckQueue.remove(msgId);
        if (ackInfo != null) {
            log.debug("消息ID: {} 已确认并从等待队列中移除", msgId);
        }
    }

    private void handleAckTimeout(Long msgId) {
        if (msgId == null) {
            log.warn("处理超时时接收到空的消息ID");
            return;
        }

        // 通过 remove 操作原子性地获取并移除消息信息
        PendingAckInfo ackInfo = pendingAckQueue.remove(msgId);
        if (ackInfo == null) {
            // 如果为 null，表示消息已经被确认，不需要处理
            return;
        }

        Channel channel = UserChannelMap.getChannelByChannelShortId(ackInfo.getUserId(), ackInfo.getChannelShortId());
        if (channel != null) {
            channel.close();
        }

        // 处理离线消息
        processOfflineMessage(ackInfo.getCode(), ackInfo.getUserId(), msgId);
    }

    private void processOfflineMessage(Integer code, Long userId, Long msgId) {
        // 根据消息类型构建离线消息事件
        OfflineMsgMqInfo offlineMsgMqInfo;
        if (IMCmdType.SYSTEM_MESSAGE.code().equals(code)) {
            offlineMsgMqInfo = new OfflineMsgMqInfo(userId, msgId, OfflineMsgMqInfo.OfflineMsgEnum.SYSTEM_OFFLINE_MESSAGE);
        } else if (IMCmdType.PRIVATE_MESSAGE.code().equals(code)) {
            offlineMsgMqInfo = new OfflineMsgMqInfo(userId, msgId, OfflineMsgMqInfo.OfflineMsgEnum.PRIVATE_OFFLINE_MESSAGE);
        } else {
            log.error("不支持的消息类型，消息ID: {}", msgId);
            throw new IllegalArgumentException("不支持的消息类型: " + code);
        }

        try {
            eventPublisher.publish(offlineMessageEvent.topic(), offlineMessageEvent.buildEventMessage(offlineMsgMqInfo));
            log.info("消息ID: {} 已超时并已发布到离线消息队列", msgId);
        } catch (Exception e) {
            log.error("消息ID: {} 发布到离线消息队列失败", msgId, e);
            // 这里可以加入重试机制或其他处理逻辑
        }
    }

    @PreDestroy
    public void destroy() {
        log.info("开始停止时间轮定时器并清理等待队列");

        try {
            WHEEL_TIMLE.stop();
            log.info("时间轮定时器已停止");
        } catch (Exception e) {
            log.error("停止时间轮定时器时发生错误", e);
        }

        try {
            pendingAckQueue.clear();
            log.info("等待队列已清理");
        } catch (Exception e) {
            log.error("清理等待队列时发生错误", e);
        }
    }
}